/**
 * Copyright (C) 2016 Newland Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.newlandframework.avatarmq.consumer;

import com.google.common.base.Joiner;
import com.newlandframework.avatarmq.core.AvatarMQAction;
import com.newlandframework.avatarmq.core.MessageIdGenerator;
import com.newlandframework.avatarmq.core.MessageSystemConfig;
import com.newlandframework.avatarmq.model.MessageType;
import com.newlandframework.avatarmq.model.RequestMessage;
import com.newlandframework.avatarmq.msg.SubscribeMessage;
import com.newlandframework.avatarmq.msg.UnSubscribeMessage;
import com.newlandframework.avatarmq.netty.MessageProcessor;

/**
 * 消息的消费者集群管理
 */
public class AvatarMQConsumer extends MessageProcessor implements AvatarMQAction {

	private boolean subscribeMessage = false;
    private boolean running = false;
    
    private ProducerMessageHook hook;
    private String brokerServerAddress;
    private String topic;
    private static final String defaultClusterId = "AvatarMQConsumerClusters";
    private String clusterId = "";
    private String consumerId = "";

    //连接的消息服务器broker的ip地址以及关注的生产过来的消息钩子
    public AvatarMQConsumer(String brokerServerAddress, String topic, ProducerMessageHook hook) {
        super(brokerServerAddress);
        this.hook = hook;
        this.brokerServerAddress = brokerServerAddress;
        this.topic = topic;
    }

    //向消息服务器broker发送取消订阅消息
    private void unRegister() {
        RequestMessage request = new RequestMessage();
        request.setMsgType(MessageType.unsubscribe);
        request.setMsgId(new MessageIdGenerator().generate());
        request.setMsgParams(new UnSubscribeMessage(consumerId));
        
        super.sendSyncMessage(request);
        super.getMessageConnectFactory().close();
        super.closeMessageConnectFactory();
        this.running = false;
    }

    //向消息服务器broker发送订阅消息
    private void register() {
        SubscribeMessage subscript = new SubscribeMessage();
        subscript.setClusterId(getClusterId());
        subscript.setTopic(topic);
        subscript.setConsumerId(consumerId);

        RequestMessage request = new RequestMessage();
        request.setMsgType(MessageType.subscribe);
        request.setMsgId(new MessageIdGenerator().generate());
        request.setMsgParams(subscript);

        super.sendAsynMessage(request);
    }
    
    
    public void init() {
        super.getMessageConnectFactory()
        	.setMessageHandle(new MessageConsumerHandler(this, new ConsumerHookMessageEvent(hook)));
        
        Joiner joiner = Joiner.on(MessageSystemConfig.MessageDelimiter).skipNulls();
        this.consumerId = joiner
        		.join(getClusterId(), topic, new MessageIdGenerator().generate());
    }
    
    private String getClusterId(){
    	return (clusterId.equals("") ? defaultClusterId : clusterId);
    }


    //连接消息服务器broker
    public void start() {
        if (this.isSubscribeMessage()) {
            super.getMessageConnectFactory().connect();
            this.register();
            this.running = true;
        }
    }

    public void receiveMode() {
        this.setSubscribeMessage(true);
    }

    public void shutdown() {
        if (running) {
            unRegister();
        }
    }

    public String getBrokerServerAddress() {
        return brokerServerAddress;
    }

    public void setBrokerServerAddress(String brokerServerAddress) {
        this.brokerServerAddress = brokerServerAddress;
    }


    public void setTopic(String topic) {
        this.topic = topic;
    }

    public boolean isSubscribeMessage() {
        return subscribeMessage;
    }

    public void setSubscribeMessage(boolean subscribeMessage) {
        this.subscribeMessage = subscribeMessage;
    }

    public void setClusterId(String clusterId) {
        this.clusterId = clusterId;
    }
}
